package com.mutouren.modules.datapush.storage;

import java.io.BufferedInputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.util.Queue;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

import org.apache.commons.lang3.StringUtils;

import com.mutouren.common.exception.ExceptionUtils;
import com.mutouren.common.utils.FileUtils;
import com.mutouren.modules.datapush.base.GradeQueue;
import com.mutouren.modules.datapush.base.PushMessage;

public class StorageHelper {
	
	public static void save(String filePath, Queue<PushMessage> queueMessage) {
		Supplier<PushMessage> supplierMessage = () -> queueMessage.poll();
		save(filePath, supplierMessage);
	}

	public static void save(String filePath, GradeQueue<PushMessage> queueMessage) {
		Supplier<PushMessage> supplierMessage = () -> queueMessage.poll();
		save(filePath, supplierMessage);		
	}
	
	private static void save(String filePath, Supplier<PushMessage> supplierMessage) {
		Function<byte[], byte[]> funcFlag = (x)-> {			
			return String.format("%d$", x.length).getBytes();
		};
				
		try (FileOutputStream writer = new FileOutputStream(filePath, true)) {
			
			PushMessage message = null;
			do {
				//message = waitingQueue.poll();
				message = supplierMessage.get();
				if (message != null) {
					byte[] buf = message.encode();
					writer.write(funcFlag.apply(buf));
					writer.write(buf);
				}
			} while (message != null);
			
		} catch(Exception e) {
			throw ExceptionUtils.doUnChecked(e);
		}
		
	}
	
	public static void load(String filePath, Queue<PushMessage> queueMessage) {
		Consumer<PushMessage> supplierMessage = (x) -> queueMessage.add(x);
		load(filePath, supplierMessage);
	}
	
	public static void load(String filePath, GradeQueue<PushMessage> queueMessage) {
		Consumer<PushMessage> supplierMessage = (x) -> queueMessage.add(x);
		load(filePath, supplierMessage);
	}
	
	private static void load(String filePath, Consumer<PushMessage> consumerMessage) {
		if (StringUtils.isBlank(filePath)) return;
		if (!FileUtils.isExist(filePath)) return;
		
		try (BufferedInputStream bf = new BufferedInputStream(new FileInputStream(new File(filePath)))) {		

			ByteArrayOutputStream out = new ByteArrayOutputStream();	
			byte[] buffer = new byte[1024];
			int len = 0;
			int dataLen = 0;
			boolean isFlag = true;
						
			do {					
				len = bf.read(buffer);
				if (len == 0) continue;
				
				for(int i = 0; i < len; i++) {
					if(isFlag) {
						if (buffer[i] == '$'){
							dataLen = Integer.valueOf(new String(out.toByteArray()));
							out.reset();
							isFlag = false;
						} else {
							out.write(buffer[i]);
						}
					} else {
						out.write(buffer[i]);
						dataLen--;
						if(dataLen == 0) {
							PushMessage message = PushMessage.decode(out.toByteArray());
							//gradeQueue.add(message);
							consumerMessage.accept(message);
															
							out.reset();
							isFlag = true;							
						}
					}
				}
								
			} while (len >= 0);
			
		}catch (Exception e) {
			throw ExceptionUtils.doUnChecked(e);
		}		
	}	
}
